package org.zjt.rxjava.lession.first;

import io.reactivex.Observable;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.schedulers.Schedulers;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.ToString;
import org.junit.Test;

import java.lang.reflect.Field;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ExecutionException;


/**
 * DESC
 *
 * @author
 * @create 2017-08-28 上午11:17
 **/
public class Scheduler {


    @Test
    public void test1() {

        Observable<Integer> objectObservable = Observable.create((emitter) -> {
            int i = new Random().nextInt(10) * 1000;
            Thread.sleep(i);
            emitter.onNext(i);
            emitter.onComplete();
        });


        objectObservable.subscribe(System.out::println);

        objectObservable.subscribe(System.out::println);

    }

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        long start = System.currentTimeMillis();
        /*//*Observable<User> users = getUser();
        Observable<UserInfo> userInfos = getUserAge();
        Future<User> userFuture = users.toFuture();
        Future<UserInfo> userInfoFuture = userInfos.toFuture();
        System.out.println(userFuture.get());
        System.out.println(userInfoFuture.get());


        System.out.println(System.currentTimeMillis() - start);*//*

        start = System.currentTimeMillis();
        Future<Object> stringFuture1 = Observable.create(
                (emitter) -> {
                    emitter.onNext(getUser("---zhang----"));
                    emitter.onComplete();
                }
        ).subscribeOn(Schedulers.computation()).toFuture();
        Future<Object> stringFuture2 = Observable.create(
                (emitter) -> {
                    emitter.onNext(getUser("---zhang2----"));
                    emitter.onComplete();
                }
        ).subscribeOn(Schedulers.computation()).toFuture();
        System.out.println(stringFuture1.get() + "     " + stringFuture2.get());
        System.out.println(System.currentTimeMillis() - start);*/

        start = System.currentTimeMillis();
        Observable<Map.Entry> objectObservable = Observable.create(
                (emitter) -> {
                    emitter.onNext(new HashMap.SimpleEntry<>("zhang1", getUser("18")));
                    emitter.onComplete();
                });

        Observable<Map.Entry> objectObservable2 = Observable.create(
                (emitter) -> {
                    emitter.onNext(new HashMap.SimpleEntry<>("zhang", getUser("20")));
                    emitter.onComplete();
                });

        Observable<Map.Entry> objectObservable3 = Observable.create(
                emitter -> {
                    emitter.onNext(new HashMap.SimpleEntry<>("zhang3", getUser("23")));
                    emitter.onComplete();
                });



        objectObservable3 = objectObservable3.subscribeOn(Schedulers.io());
        objectObservable2 = objectObservable2.subscribeOn(Schedulers.io());
        objectObservable = objectObservable.subscribeOn(Schedulers.io());



        Observable.merge(Arrays.asList(objectObservable, objectObservable2, objectObservable3))
                .blockingSubscribe((Map.Entry a) -> {
                    Object key = a.getKey();
                    Object value = a.getValue();
                    System.out.println(a.getClass()+" "+a);
                });

        /*Observable.zip(Arrays.asList(objectObservable, objectObservable2, objectObservable3), (Function) result -> {
            System.out.println(result);
            return result;
        }).blockingSubscribe(System.out::println);*/
        System.out.println(System.currentTimeMillis() - start);


        /*Observable.zip(users, userInfos, (user, userInfo) -> {
//            System.out.println(user + "  " + userInfo);
            return UserInfo2Map(user, userInfo);
        }).blockingSubscribe(System.out::println);
        userInfos.subscribe(System.out::println);


        Flowable.just("Hello world").subscribe(System.out::println);


        *//**
         *  即使发生异常，只会中断该流的方法。下面继续执行。
         *//*
        Flowable.fromCallable(() -> {
            Thread.sleep(4000);
            int i = 1 / 0;
            return "Done";
        }).map(a -> {
            System.out.println("  map  " + a);
            return a;
        }).subscribeOn(Schedulers.io())
                .subscribeOn(Schedulers.single())
                .blockingSubscribe(System.out::println, Throwable::printStackTrace);

        System.out.println((System.currentTimeMillis() - start) / 1000);


        Flowable.range(1, 10).flatMap(v ->
                Flowable.just(v).subscribeOn(Schedulers.computation()).map(Math::sqrt)
        ).blockingSubscribe(System.out::println);
*/

    }


    public static Observable<User> getUser() {
        return Observable.create((ObservableOnSubscribe<User>) observableEmitter -> {
            Thread.sleep(4 * 1000);
            observableEmitter.onNext(new User(1, "zhangsan"));
            /*observableEmitter.onNext(new User(2, "lisi"));*/
            observableEmitter.onComplete();
        }).subscribeOn(Schedulers.computation());
    }

    public static String getUser(String name) throws InterruptedException {
        Thread.sleep(500);
        return name.toUpperCase();
    }


    public static Map<String, Object> UserInfo2Map(Object... objects) throws IllegalAccessException {
        Map<String, Object> data = new HashMap<>();
        for (Object o : Arrays.asList(objects)) {
            for (Field field : o.getClass().getDeclaredFields()) {
                field.setAccessible(true);
//                System.out.println(field.getName() +"             " +field.get(o).toString());

                data.put(field.getName(), field.get(o));
            }
        }
        return data;
    }


    public static Observable<UserInfo> getUserAge() {
        return Observable.create((ObservableOnSubscribe<UserInfo>) observableEmitter -> {
            Thread.sleep(4 * 1000);
            observableEmitter.onNext(new UserInfo(1, 1, "shanxi"));
/*
            observableEmitter.onNext(new UserInfo(2, 2, "beijing"));
*/
            observableEmitter.onComplete();
        }).subscribeOn(Schedulers.computation());
    }


}

@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
class User {
    private Integer id;
    private String name;
}


@Data
@AllArgsConstructor
@NoArgsConstructor
@ToString
class UserInfo {
    private Integer id;
    private Integer userId;
    private String address;

}

